//Team 4 - Gold Coast
//Duy Kha Nguyen 598240 
//Rongkai Zhou 562946 
//Zhangyue Cheng 548692
//Binzan Liang 591911
//Wei Liu 572322
package twitter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;
import org.lightcouch.CouchDbClient;

import com.fasterxml.jackson.databind.ObjectMapper;

public class StoreTweetTask implements Runnable {

    private static final Logger logger = Logger.getLogger(StoreTweetTask.class);

    private final BlockingQueue<String> tweetQueue;

    private final CouchDbClient couchDb;

    private final ObjectMapper jsonParser;

    public StoreTweetTask(BlockingQueue<String> tweetQueue, CouchDbClient couchDb, ObjectMapper jsonParser) {
        super();
        this.tweetQueue = tweetQueue;
        this.couchDb = couchDb;
        this.jsonParser = jsonParser;
    }

    public void run() {
        final List<String> rawTweets = new ArrayList<String>(1024);
        final List<Map> tweets = new ArrayList<Map>(1024);
        while (true) {
            tweetQueue.drainTo(rawTweets, 1024);
            for (final String rawTweet : rawTweets) {
                try {
                    final Map tweet = jsonParser.readValue(rawTweet, Map.class);
                    tweet.put("_id", tweet.get("id_str"));
                    tweet.remove("id_str");
                    tweets.add(tweet);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (tweets.size() > 0) {
                couchDb.bulk(tweets, false);
                rawTweets.clear();
                tweets.clear();
            } else {
                try {
                    logger.info("sleeping");
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
